package com.zhang.third.day10;

import com.zhang.third.utils.OrderEvent;
import com.zhang.third.utils.ReceiptEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @title: 实时对账
 * @author: zhang
 * @date: 2022/4/15 07:03
 */
public class OrderReceipt2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取订单数据
        SingleOutputStreamOperator<OrderEvent> orderDS = env
                .readTextFile("/Users/apple/IdeaProjects/flink_1.13/src/main/resources/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] arr = value.split(",");
                        return new OrderEvent(
                                arr[0], arr[1], arr[2],
                                Long.parseLong(arr[3]) * 1000L
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                                    @Override
                                    public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .filter(r -> r.type.equals("pay"));

        // 读取到账数据流
        SingleOutputStreamOperator<ReceiptEvent> payDS = env
                .readTextFile("/Users/apple/IdeaProjects/flink_1.13/src/main/resources/ReceiptLog.csv")
                .map(new MapFunction<String, ReceiptEvent>() {
                    @Override
                    public ReceiptEvent map(String value) throws Exception {
                        String[] arr = value.split(",");
                        return new ReceiptEvent(
                                arr[0], arr[1],
                                Long.parseLong(arr[2]) * 1000L
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ReceiptEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<ReceiptEvent>() {
                                    @Override
                                    public long extractTimestamp(ReceiptEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        // 实时对账逻辑
        SingleOutputStreamOperator<String> result = orderDS
                .keyBy(r -> r.payId)
                .connect(payDS.keyBy(r -> r.payId))
                .process(new MatchOrder());

        result.print("主流");
        result.getSideOutput(new OutputTag<String>("no-pay") {
        }).print("侧输出流");
        result.getSideOutput(new OutputTag<String>("no-order") {
        }).print("侧输出流");


        env.execute();
    }

    public static class MatchOrder extends CoProcessFunction<OrderEvent, ReceiptEvent, String> {
        private ValueState<OrderEvent> orderState;
        private ValueState<ReceiptEvent> payState;
        private ValueState<Long> timerState;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderState = getRuntimeContext().getState(
                    new ValueStateDescriptor<OrderEvent>(
                            "order",
                            Types.POJO(OrderEvent.class)
                    )
            );

            payState = getRuntimeContext().getState(
                    new ValueStateDescriptor<ReceiptEvent>(
                            "pay",
                            Types.POJO(ReceiptEvent.class)
                    )
            );

            timerState = getRuntimeContext().getState(
                    new ValueStateDescriptor<Long>(
                            "timer",
                            Types.LONG
                    )
            );
        }

        @Override
        public void processElement1(OrderEvent value, CoProcessFunction<OrderEvent, ReceiptEvent, String>.Context ctx, Collector<String> out) throws Exception {
            if (payState.value() != null) {
                out.collect(value.payId + ": 对账成功，pay事件先到达");
                ctx.timerService().deleteEventTimeTimer(timerState.value());
                timerState.clear();
                payState.clear();
            } else {
                orderState.update(value);
                long tenSecLater = value.ts + 10 * 1000L;
                ctx.timerService().registerEventTimeTimer(tenSecLater);
                timerState.update(tenSecLater);
            }
        }

        @Override
        public void processElement2(ReceiptEvent value, CoProcessFunction<OrderEvent, ReceiptEvent, String>.Context ctx, Collector<String> out) throws Exception {
            if (orderState.value() != null) {
                out.collect(value.payId + ": 对账成功，order事件先到达");
                ctx.timerService().deleteEventTimeTimer(timerState.value());
                timerState.clear();
                orderState.clear();
            } else {
                payState.update(value);
                long fiveSecLater = value.ts + 10 * 1000L;
                ctx.timerService().registerEventTimeTimer(fiveSecLater);
                timerState.update(fiveSecLater);
            }
        }

        @Override
        public void onTimer(long timestamp, CoProcessFunction<OrderEvent, ReceiptEvent, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            if (orderState.value() != null) {
                ctx.output(new OutputTag<String>("no-pay") {
                }, orderState.value().payId + ": 没有被支付");
                orderState.clear();
            } else {
                ctx.output(new OutputTag<String>("no-order") {
                }, payState.value().payId + ": 没有被创建");
                payState.clear();
            }
            timerState.clear();
        }
    }
}
